# Unspool.pm
# Copyright (C) 2006, 2007  Stephane Alnet
#
# This program is free software; you can redistribute it and/or
# modify it under the terms of the GNU General Public License
# as published by the Free Software Foundation; either version 3
# of the License, or (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program.  If not, see <http://www.gnu.org/licenses/>.
# 

use strict; use warnings;

package CCN::IPC::Unspool;
use base qw(CCN::Class);

use CCN::Daemon;
use CCN::IPC::SingleInstance;

sub initialize()
{
    our $self = shift;

    # Preparatory code
    $self->{daemon} = new CCN::Daemon 'Unspool';

    $self->{single} = new CCN::IPC::SingleInstance($self->here(),$self->cmd());

    $self->{pending} = 0;

    sub quit_unspool()
    {
        $self->info('Request to stop unspool.');
        $self->info("There are still $self->{pending} children, exiting anyway")
            if $self->{pending} > 0;
        $self->{single}->stop;
    }

    sub restart_unspool()
    {
        $self->info('Request to restart Unspool.');
        $self->info("There are still $self->{pending} children, exiting anyway")
            if $self->{pending} > 0;
        undef $self->{daemon};
        $self->{single}->restart;
    }

    sub ping()
    {
        $self->info('Ping');
        return 'Pong';
    }

    $self->register_commands(

        # Internal (no params)
        'quit-unspool'      => ['internal', \&quit_unspool],
        'restart-unspool'   => ['internal', \&restart_unspool],
        'ping'              => ['internal', \&ping],

    );
}

sub register_commands
{
    my $self = shift;
    my %commands = @_;
    $self->{commands}->{(keys %commands)} = (values %commands);
}

sub local_unspool()
{
    my $self = shift;

    # No more than 10 requests pending at any time.
    if( $self->{pending} > 10 )
    {
        return 'Too many pending requests';
    }

    my $info = $self->spool->message_store->get_next($self->spool->get_local_pk());

    # Important: MUST return something.
    return 'Nothing to process' if not defined $info;

    # Clear the message even if we failed -- this allows us to advance through the messages.
    $self->spool->message_store->clear($info->{pk});

    my $what = $info->{content};

    return "Invalid request"
        if not defined $what or $what eq '';

    # Process the request
    chomp $what;
    my @what = split(' ',$what);
    my $cmd = shift @what;

    return "Invalid command $what"
        unless exists $self->{commands}->{$cmd};

    # Internal commands
    if( $self->{commands}->{$cmd}->[0] eq 'internal' )
    {
        return $self->{commands}->{$cmd}->[1]->(@what);
    }

    # External commands
    $SIG{CHLD} = \&unthrottle;
    my $fork = fork();
    return 'Fork failed' if not defined $fork;

    # Child process
    if( $fork == 0 )
    {
        # Restore the CHLD handler so that the subprocesses can fork if they need to.
        $SIG{CHLD} = $self->{CHLD_HANDLER};
        # Child process
        $self->info('Executing '.join(' ',@{$self->{commands}->{$cmd}}, @what));
        # close(STDIN); # This is done by CCN::Daemon
        # close(STDOUT);
        # close(STDERR);
        undef $self->{daemon};

        # A Perl module that offers a "run" command
        if( $self->{commands}->{$cmd}->[0] eq 'perl' )
        {
            my $module = $self->{commands}->{$cmd}->[1];
            our $d = new CCN::Syslogger $module;
            eval "use ${module}; \$${module}::D = \$d; ${module}::run(\@what)";
            $self->error("eval: $@") if $@;
        }
        # Any executable
        else
        {
            $ENV{PATH} = '/usr/sbin:/usr/bin:/sbin:/bin';
            exec @{$self->{commands}->{$cmd}}, @what;
        }
        exit 0;
    }

    # Parent process
    $self->{pending}++;
    $self->{throttle}++;

    # No more than 5 requests forked by period.
    if( $self->{throttle} > 5 )
    {
        return 'Too many requests';
    }
    return '';
}

sub run()
{
    our $self = shift;

    # It's still running if it was last seen no less than one minutes ago.
    $self->log_for('info','Unspool already running.'), exit 0
        if defined $self->{single}->last_touched and $self->{single}->last_touched > time() - 90;

    # /Preparatory code

    # Number of pending requests.
    $self->{pending} = 0;

    # Save the original "CHLD" handler
    $self->{CHLD_HANDLER} = $SIG{CHLD};

    sub unthrottle()
    {
        my $waitedpid = wait;
        $self->{pending}-- if $self->{pending} > 0;
        $SIG{CHLD} = \&unthrottle;
    }

    # Number of requests over the last second
    $self->{throttle} = 0;

    # This is called from crontab every 10 minutes.
    # Only one should be running at any time.

    # Stay up

    # On Linux, have to use the method described in perlipc:
    use POSIX;
    my $sigset = POSIX::SigSet->new();
    my $action = POSIX::SigAction->new($self->{commands}->{'restart-unspool'}->[1],
                                       $sigset,
                                       &POSIX::SA_NODEFER);
    POSIX::sigaction(&POSIX::SIGHUP, $action);
    
    $SIG{TERM} = $self->{commands}->{'quit-unspool'}->[1];
    $SIG{INT}  = $self->{commands}->{'quit-unspool'}->[1];

    $self->{single}->touch();
    $self->info( 'Started.' );
    
    our $delay = 1;

    # STDOUT and STDERR should be redirected (for the children processes)
    while(1)
    {
        my $exit = $self->local_unspool();

        if($exit)
        {
            if( $exit eq 'Nothing to process' )
            {
                $delay = 1;
            }
            else
            {
                $self->log_for('info',$exit);
            }
            $delay *= 2 if $delay <= 4 and $exit eq 'Too many pending requests';
            sleep($delay);
            $self->{throttle} = 0;
        }
        # Otherwise: processed OK, no throttling

        # Re-touch every 20 seconds at most
        if( not defined $self->{single}->last_touched or $self->{single}->last_touched < time() - 20 )
        {
            exit 1 unless $self->{single}->touch();
        }
    }
    # NEVER REACHED
}

1;